package com.example.mq.mqserver.datacenter;

import com.example.mq.common.tool.BinaryTool;
import com.example.mq.common.exception.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/*
 * 通过这个类, 来针对硬盘上的消息进行管理
 */
@Slf4j
public class MessageFileManager {

    // 内部类, 来表示该队列的统计信息
    static public class Stat {
        public int totalCount;  // 总消息数量
        public int validCount;  // 有效消息数量
    }

    // 定义路径和文件名常量
    private static final String BASE_DIR = "data";
    private static final String DATA_FILE_NAME = "queue_data.dat";
    private static final String DATA_FILE_NAME_NEW = "queue_data_new.dat";

    private static final String STAT_FILE_NAME = "queue_stat.dat";

    private static final int MESSAGE_SIZE_NUM = 4;   // 记录消息大小的内存大小，4字节

    public void init() {
        // 暂时不需要，以备后续扩展
    }

    // 获取指定队列对应的消息文件所在目录路径
    private String getQueueDir(String queueName) {
        return BASE_DIR + File.separator + queueName;
    }

    // 获取队列消息数据文件路径
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + File.separator + DATA_FILE_NAME;
    }

    // 获取队列消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + File.separator + STAT_FILE_NAME;
    }

    // 获取迁移文件的文件路径
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + File.separator + DATA_FILE_NAME_NEW;
    }

    // 从消息统计文件中读取统计数据
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (Scanner scanner = new Scanner(new FileInputStream(getQueueStatPath(queueName)))) {
            // 读取总消息数和有效消息数
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
        } catch (IOException e) {
            log.error(e.getMessage());
        }
        return stat;
    }

    // 将统计数据写入到消息统计文件中
    private void writeStat(String queueName, Stat stat) {
        try (PrintWriter printWriter = new PrintWriter(new FileOutputStream(getQueueStatPath(queueName)))) {
            // 写入总消息数和有效消息数，使用制表符分隔
            printWriter.println(stat.totalCount + "\t" + stat.validCount);
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    // 创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName)); // 获取队列目录的路径
        if (!baseDir.exists()) { // 检查目录是否存在
            if (!baseDir.mkdirs()) { // 如果目录创建失败
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath()); // 抛出异常
            }
        }

        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName)); // 获取队列数据文件的路径
        if (!queueDataFile.exists()) { // 检查数据文件是否存在
            if (!queueDataFile.createNewFile()) { // 如果文件创建失败
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }

        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName)); // 获取消息统计文件的路径
        if (!queueStatFile.exists()) { // 检查统计文件是否存在
            if (!queueStatFile.createNewFile()) { // 如果文件创建失败
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }

        // 4. 给消息统计文件设定初始值
        Stat stat = new Stat(); // 创建一个统计对象
        stat.totalCount = 0; // 初始化消息总数为0
        stat.validCount = 0; // 初始化有效消息数为0
        writeStat(queueName, stat); // 将统计信息写入文件
    }


    // 删除队列的目录和文件.
    public void destroyQueueFiles(String queueName) throws IOException {
        // 获取各个文件和目录对象
        File queueDataFile = new File(getQueueDataPath(queueName));
        File queueStatFile = new File(getQueueStatPath(queueName));
        File baseDir = new File(getQueueDir(queueName));

        // 检查并删除文件和目录
        boolean isDeleted = queueDataFile.delete() && queueStatFile.delete() && baseDir.delete();

        // 如果有任何一个删除失败，则抛出异常
        if (!isDeleted) {
            throw new IOException("删除队列目录和文件失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }


    // 检查队列的目录和文件是否存在.
    // 比如后续有生产者给 broker server 生产消息了, 这个消息就可能需要记录到文件上(取决于消息是否要持久化)
    public boolean checkFilesExits(String queueName) {
        // 判定队列的数据文件和统计文件是否都存在!!
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        return queueStatFile.exists();
    }

    // 这个方法用来把一个新的消息, 放到队列对应的文件中.
    // queue 表示要把消息写入的队列. message 则是要写的消息.
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1. 检查队列对应的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName=" + queue.getName());
        }

        // 2. 将 Message 对象序列化为字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);

        synchronized (queue) {
            // 3. 计算 Message 的 offsetBeg 和 offsetEnd
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            long fileLength = queueDataFile.length();
            message.setOffsetBeg(fileLength + MESSAGE_SIZE_NUM); // 当前文件长度 + 4 (长度字段)
            message.setOffsetEnd(fileLength + MESSAGE_SIZE_NUM + messageBinary.length); // 当前文件长度 + 4 + 消息长度

            // 4. 追加写入消息到文件末尾
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                dataOutputStream.writeInt(messageBinary.length); // 写入消息长度
                dataOutputStream.write(messageBinary); // 写入消息内容
            }

            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    // 删除某段消息（逻辑删除）
    // 1. 先把文件中的这一段数据, 读出来, 还原回 Message 对象;
    // 2. 把 isValid 改成 0;
    // 3. 把上述数据重新写回到文件.
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 从文件中读取对应的消息数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());  // 将文件指针移动到删除消息起点
                randomAccessFile.readFully(bufferSrc);  // 读取到buf中

                // 2. 反序列化为 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);

                // 3. 将 isValid 设置为 0 (无效)
                diskMessage.setIsValid((byte) 0x0);

                // 4. 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg()); // 恢复到原位置
                randomAccessFile.write(bufferDest);
            }

            // 更新统计文件，减少有效消息计数
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }

    // 从文件中读取所有消息内容并加载到内存中的链表中
    // 使用这个方法, 从文件中, 读取出所有的消息内容, 加载到内存中(具体来说是放到一个链表里)
    // 这个方法, 准备在程序启动的时候, 进行调用.
    // 在程序启动时调用, 不涉及多线程操作文件.
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();

        try (DataInputStream dataInputStream = new DataInputStream(new FileInputStream(getQueueDataPath(queueName)))) {
            long currentOffset = 0; // 当前文件光标

            while (true) {
                try {
                    // 1. 读取消息长度
                    int messageSize = dataInputStream.readInt();    // 读一个字节

                    // 2. 读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }

                    // 3. 反序列化消息对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);

                    // 4. 过滤无效消息
                    if (message.getIsValid() != 0x1) {
                        currentOffset += (MESSAGE_SIZE_NUM + messageSize);
                        continue;
                    }

                    // 5. 记录消息偏移量并添加到链表
                    message.setOffsetBeg(MESSAGE_SIZE_NUM + currentOffset);
                    message.setOffsetEnd(MESSAGE_SIZE_NUM + currentOffset + messageSize);
                    currentOffset += (MESSAGE_SIZE_NUM + messageSize);
                    messages.add(message);

                } catch (EOFException e) {
                    // 正常的结束条件：文件读取到末尾
                    log.info("[MessageFileManager] 恢复 Message 数据完成!");
                    break;
                }
            }
        }

        return messages;
    }


    // 检查是否需要对指定队列的消息数据文件进行 GC
    public boolean checkGC(String queueName) {
        // 从消息统计文件中读取消息统计信息
        Stat stat = readStat(queueName);

        // 如果总消息数大于 2000 且 有效消息数 / 总消息数 < 0.5，则需要进行 GC
        return stat.totalCount > 2000 && (double) stat.validCount / stat.totalCount < 0.5;
    }

    // 执行消息数据文件的垃圾回收操作，使用复制算法来完成.
    // 创建一个新的文件,把有效消息都读出来, 写到新的文件中.
    // 删除旧的文件, 再把新的文件改名回旧的文件名
    // 同时要记得更新消息统计文件.
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue) {
            long gcBeg = System.currentTimeMillis(); // 记录开始时间

            // 1. 创建新的数据文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                throw new MqException("[MessageFileManager] gc 时发现队列的 queue_data_new 已经存在! queueName=" + queue.getName());
            }
            if (!queueDataNewFile.createNewFile()) {
                throw new MqException("[MessageFileManager] 创建新文件失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }

            // 2. 读取所有有效消息
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            // 3. 将有效消息写入新文件
            try (DataOutputStream dataOutputStream = new DataOutputStream(new FileOutputStream(queueDataNewFile))) {
                for (Message message : messages) {
                    byte[] buffer = BinaryTool.toBytes(message);
                    dataOutputStream.writeInt(buffer.length);
                    dataOutputStream.write(buffer);
                }
            }

            // 4. 删除旧文件并重命名新文件
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            if (!queueDataOldFile.delete()) {
                throw new MqException("[MessageFileManager] 删除旧文件失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            if (!queueDataNewFile.renameTo(queueDataOldFile)) {
                throw new MqException("[MessageFileManager] 重命名新文件失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath()
                        + ", queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }

            // 5. 更新统计信息
            Stat stat = new Stat();
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            long gcEnd = System.currentTimeMillis(); // 记录结束时间
            log.info("[MessageFileManager] gc 执行完毕! queueName={}, time={}ms", queue.getName(), gcEnd - gcBeg);
        }
    }
}
